package org.myorg.v6;
import java.io.IOException;
import java.net.ConnectException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.Timer;
import java.util.Vector;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.ObjectWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.Server;
import org.myorg.v4.FPeopleWritable4;

public class TransImpl implements TransProtocol {

	public String ID = new String();
	public int port = 8888;
	public int NetWorkType = 0;
	public int MapNum = 0;
	public InetSocketAddress myip = null;
	public InetSocketAddress monitor = null;

	public Vector<InetSocketAddress> address = new Vector<InetSocketAddress>();

	private Server server;
	private TransProtocol proxy;
	private int[] lock = new int[0];
	private boolean flag = false;
	private int[] addr_lock = new int[0];
	//private boolean addr_flag = false;
	private Vector<FPeopleWritable4> tp = new Vector<FPeopleWritable4>();
	private int count = 0; 
	private long period = 0;
	private boolean too_long = true;
	private long WaitingTime = 30*1000;

	private static final Log LOG = LogFactory.getLog(TransImpl.class);

	private Timer timer = new Timer(true); 

	public TransImpl(String ID, int mode, int MapNum, InetSocketAddress monitor, int period) {

		this.ID = ID;
		this.NetWorkType = mode;
		this.MapNum = MapNum;
		this.monitor = monitor;
		this.period = period;

	}

	public void start() {
		StartServer();
		register();
		timer.schedule(
				new java.util.TimerTask() { 
					public void run() {
						register();
					} 
				}, 
				this.period, 
				this.period
		);

	}

	public void stop() {
		timer.cancel();
		server.stop();
	}


	public void StartServer() {

		int count = 0;
		boolean running = true;

		while (count < 1000 && running) {
			this.port = (int)Math.round(Math.random()*(65536-1024) + 1024);
			try {
				myip = new InetSocketAddress(InetAddress.getLocalHost().getHostAddress(),port);
			} catch (UnknownHostException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
			try {
				server = RPC.getServer(this, myip.getAddress().getHostAddress(), port, new Configuration());
				server.start();
				running = false;
			} catch (UnknownHostException e) {
				count++;
				e.printStackTrace();

			} catch (IOException e) {
				count++;
				LOG.debug(myip + " has been used ");
				e.printStackTrace();
			} 
		}


	}

	public void register() {
		// TODO Auto-generated method stub

		RegisterProtocol rp = null;

		//LOG.debug("begin get proxy in register");

		//InetSocketAddress test = new InetSocketAddress("172.16.18.79", 8888);
		try {
			rp = (RegisterProtocol) RPC.waitForProxy(RegisterProtocol.class, 1, monitor, new Configuration());
		} catch (IOException e) {
			e.printStackTrace();
		}

		//LOG.debug("Begin register " + myip);

		InetSocketAddressWriatble[] IPs = rp.register(ID, myip.getAddress().getHostAddress(), port, period);

		RPC.stopProxy(rp);


		synchronized (addr_lock) {
			address.clear();
			for (InetSocketAddressWriatble ip:IPs) {
				address.add(new InetSocketAddress(ip.addr,ip.port));
			}
			addr_lock.notifyAll();
		}

		//LOG.debug("end register, the address size is " + address.size());

	}

	@Override
	public void transmit(FPeopleWritable4 o) {
		// TODO Auto-generated method stub
		//LOG.debug("Now is in transmit");
		/*synchronized (data) {
           this.data = o.get();
           this.orign = (InetSocketAddress)me.get();
           data.notifyAll();
       }*/
		synchronized (lock) {
			tp.add(o);
			flag = true;
			too_long = false;
			lock.notify();
		}

		//System.out.println("will leave transmit");


	}

	public void send(FPeopleWritable4 o) throws IOException { 
		LOG.debug("Enter new send");
		IOException ioe;
		
		InetSocketAddress target = null; 
		
		synchronized (addr_lock) {
			//LOG.debug("get addr_lock");
			while (address.isEmpty()) {
				LOG.debug("address is empty!");
				try {
					addr_lock.wait();
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
			target = address.firstElement();
		}
		
		send(o,target);
	}

	public void send(FPeopleWritable4 o, InetSocketAddress target)throws IOException {
		//LOG.debug("Enter send");
		//InetSocketAddress test = new InetSocketAddress("172.16.18.79", 8888);
		IOException ioe;
		try {
			proxy = (TransProtocol) RPC.getProxy(TransProtocol.class, 1, target, new Configuration());
		} catch(ConnectException se) {  // namenode has not been started
			LOG.debug("Server at " + monitor + " not available yet, Zzzzz...");
			ioe = se;
			throw ioe;
		} catch(SocketTimeoutException te) {  // namenode is busy
			LOG.debug("Problem connecting to server: " + monitor);
			ioe = te;
			throw ioe;
		}
		LOG.debug("Begin transmit to" + target);
		proxy.transmit(o);
		RPC.stopProxy(proxy);
	}

	public FPeopleWritable4 Wait4Data() {
		// TODO Auto-generated method stub

		LOG.debug("in Wait4Data");
		FPeopleWritable4 rt = null;

		synchronized(lock){
			while (tp.isEmpty()) {
				LOG.debug("Will be waiting");
				try {                                   
					lock.wait(WaitingTime); 
				} catch (InterruptedException e) {      
				} 
				if (too_long) {
					break;
				}
			}
			LOG.debug("receive data");
			if (!tp.isEmpty()) {
				rt = new FPeopleWritable4(tp.remove(0));
			}

			flag = false;
		}
		if (too_long) {
			LOG.info("In wait4data, wake up by time");
		}
		too_long = true;
		return rt;
	}

	public LinkedList<FPeopleWritable4> Wait4AllData() {
		// TODO Auto-generated method stub
		LOG.debug(" enter Wait4AllData");
		LinkedList<FPeopleWritable4> alldata = new LinkedList<FPeopleWritable4>();

		while (count  < address.size() - 1) {
			LOG.debug( "count is " + count);
			synchronized(lock){
				LOG.debug("get lock");
				while (tp.isEmpty()) {
					LOG.debug("wait for waking up");
					try {                                   
						lock.wait(WaitingTime);
					} catch (InterruptedException e) {      
					}
					//testData = new int[10];
					LOG.debug("up, too_long is " + too_long);
					if (too_long) {
						break;
					}
				}
				LOG.debug("wake up");
				if (!tp.isEmpty()) {
					LOG.debug("wake up by tp, the tp is not empty, count is " + count);
					alldata.add(tp.remove(0));

					too_long = true;
					flag = false;
				} else {
					LOG.info("the time is too long for wait all data, count is " + count);
					flag = false;
					break;
				}
				count++;
			}
		}
		count = 0;
		return alldata;
	}

	public void send2All(FPeopleWritable4 p) {
		// TODO Auto-generated method stub
		LOG.debug("send to all, the address size is " + address.size());
		synchronized (addr_lock) {
			while (address.isEmpty()) {
				LOG.debug("address is empty!");
				try {
					addr_lock.wait();
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
			for (int i = 0; i < address.size(); i++) {
				
				if (!myip.equals(address.get(i))){
					try {
						send(p, address.get(i));
					} catch (IOException e) {
						// TODO Auto-generated catch block
						LOG.warn("cann't send the data");
						e.printStackTrace();
					}
				}
			}
		}

	}

	public boolean isMaster() {
		if (address.isEmpty()) {
			return true;
		} else {
			return myip.equals(address.firstElement());
		}
	}

	@Override
	public long getProtocolVersion(String protocol, long clientVersion)
	throws IOException {
		// TODO Auto-generated method stub
		return 1;
	}	


}
